package com.millstein;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class JoinTest2 {

    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));

        tableEnv.executeSql(
                "create table t1 (\n" +
                "    id string,\n" +
                "    name string,\n" +
                "    primary key(id) not enforced\n" +
                ") with (\n" +
                "    'connector' = 'upsert-kafka',\n" +
                "    'properties.bootstrap.servers' = 'hadoop102:9092',\n" +
                "    'properties.group.id' = 'tsing',\n" +
                "    'topic' = 's1',\n" +
                "    'key.format' = 'csv',\n" +
                "    'value.format' = 'csv'\n" +
                ")"
        );

//        tableEnv.executeSql(
//                "create table t2 (\n" +
//                "    id string,\n" +
//                "    age int,\n" +
//                "    primary key(id) not enforced\n" +
//                ") with (\n" +
//                "    'connector' = 'upsert-kafka',\n" +
//                "    'properties.bootstrap.servers' = 'hadoop102:9092',\n" +
//                "    'properties.group.id' = 'tsing',\n" +
//                "    'topic' = 's2',\n" +
//                "    'key.format' = 'csv',\n" +
//                "    'value.format' = 'csv'\n" +
//                ")"
//        );

//        Table table = tableEnv.sqlQuery(
//                "select t1.id, t1.name, t2.age from t1 left join t2 on t1.id = t2.id"
//        );

        Table table = tableEnv.sqlQuery(
                "select * from t1"
        );

        table.execute().print();
    }
}
